前幾天分別談了 Embedding、向量資料庫選型與相似度搜尋原理,今天要把這些拼圖組裝起來,做一個從「輸入問題」到「拿到檢索結果」的最小可行版本(MVP),為 Indexing 與 Retrieval 這兩個階段做一個小結。
輸入一個自然語言問題,系統回傳資料庫中最相關的 K 筆 chunk 內容。
import psycopg2
import openai
DB_CONFIG = {
"host": "localhost",
"dbname": "rag_demo",
"user": "postgres",
"password": "your_password",
}
def get_query_embedding(question, model="text-embedding-3-small"):
response = openai.embeddings.create(input=question, model=model)
return response.data[0].embedding
def search_similar_chunks(question, top_k=5):
query_embedding = get_query_embedding(question)
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute(
"""
SELECT content, metadata, embedding <=>%s::vector AS distance
FROM chunks
ORDER BY distance ASC
LIMIT%s
""",
(query_embedding, top_k)
)
results = cur.fetchall()
cur.close()
conn.close()
return [
{"content": r[0], "metadata": r[1], "distance": r[2]}
for r in results
]
if __name__ == "__main__":
question = "RAG 系統的三大環節是什麼?"
results = search_similar_chunks(question, top_k=3)
for i, r in enumerate(results, 1):
print(f"[{i}] 距離:{r['distance']:.4f}")
print(f" 來源:{r['metadata'].get('source')}")
print(f" 內容:{r['content'][:100]}...")
print()
get_query_embedding:把使用者的問題,用跟 Indexing 階段相同的 Embedding 模型轉成向量——這一點很重要,查詢向量與資料庫向量必須來自同一個模型,否則向量空間不一致,比對出來的距離沒有意義search_similar_chunks:組出 SQL 查詢,利用 pgvector 的 <=> 運算子(餘弦距離)排序,取出距離最小(最相似)的 K 筆結果跑完這支程式後,建議做一次簡單的「肉眼驗收」:
這個版本只做了最基礎的 Top-K 向量搜尋,還沒有加入:
這些都是接下來幾天(Day 15 開始)要陸續補上的優化。先有一個能跑的最小版本,之後每次優化都可以直接比較「加了這個機制之後,結果有沒有變好」。
從 Day 5 的資料前處理,一路走到今天可以真正查詢的向量資料庫,Indexing 與基礎 Retrieval 的骨架已經完成。接下來一週,我們會把焦點放在如何讓檢索結果更精準——這往往是決定一套 RAG 系統實用與否的關鍵。